package com.bbx.flink.table_api_or_sql;

import com.bbx.flink.demo.entity.Student;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 1、table 转换dataStream
 * 2、dataStream 转换 table
 * 3、查看执行计划  Table.explain()
 *
 *
 */
@Slf4j
public class IntegrateDataStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //create dataStream
        SingleOutputStreamOperator<Student> dataStream = env.socketTextStream("192.168.10.131", 10003)
                .map(i -> {
            String[] split = i.split(",");
          return   new Student(split[0], split[1], Integer.parseInt(split[2]), split[3]);
        });

        //创建零时 view
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //创建零时 view
//        tableEnv.useCatalog("test_catalog");
//        tableEnv.useDatabase("test_database");
//        tableEnv.createTemporaryView("student",dataStream,$("id"),$("name"));

        //dataStream 转换称 table
        Table table = tableEnv.fromDataStream(dataStream);
        Table filterTable = table.filter($("id").isGreater("3"));

        String explain = filterTable.explain();
        log.info("explain[{}]",explain);

        //table 转换成DataStream
        DataStream<Student> studentDataStream = tableEnv.toAppendStream(filterTable, Student.class);
        studentDataStream.print("studentStream");
        env.execute();


    }



}
